package mq

import (
	"encoding/json"
	"fmt"
	"reflect"
	"time"

	"github.com/astaxie/beego/httplib"
	"github.com/nsqio/go-nsq"
	"yingsheng.com/go-api-server/config"
	"yingsheng.com/go-api-server/errors"
	"yingsheng.com/go-api-server/log"
)

func init() {
	pv := reflect.ValueOf(&mqNsqProducer{})
	pt := reflect.Indirect(pv).Type()

	cv := reflect.ValueOf(&mqNsqConsumer{})
	ct := reflect.Indirect(cv).Type()

	producers["nsq"] = pt
	consumers["nsq"] = ct
}

type mqNsqProducer struct {
	isInit   bool
	topic    string
	producer *nsq.Producer
}

func (p *mqNsqProducer) Init(topic string) (err error) {
	if len(topic) == 0 {
		err = errors.New(`NSQ: Producer Init error "topic cannot be empty"`)
		return
	}

	addr, err := p.getService()
	if err != nil {
		return
	}

	nconf := nsq.NewConfig()
	np, err := nsq.NewProducer(addr, nconf)
	if err != nil {
		err = fmt.Errorf("NSQ: Producer Init error %q", err)
		return
	}

	np.SetLogger(log.Logger, nsq.LogLevelWarning)
	p.topic = topic
	p.producer = np
	p.isInit = true
	return
}

func (p *mqNsqProducer) Publish(msg []byte) (err error) {
	if !p.isInit {
		err = errors.New(`NSQ：Publish error "must first call the Init func"`)
		return
	}

	err = p.producer.Publish(p.topic, msg)
	if err != nil {
		err = fmt.Errorf("NSQ: Publish error %q", err)
	}

	return
}

func (p *mqNsqProducer) Stop() (err error) {
	if !p.isInit {
		err = errors.New(`NSQ：Producer Stop error "must first call the Init func"`)
		return
	}

	p.producer.Stop()
	return
}

func (p *mqNsqProducer) getService() (addr string, err error) {
	lookupAddr := fmt.Sprintf("http://%s:%v/nodes", config.Configer.ApiConf.MqServer, config.Configer.ApiConf.MqPort)

	req := httplib.NewBeegoRequest(lookupAddr, "GET")
	req.SetTimeout(5*time.Second, 5*time.Second)
	//req.Param("topic", p.topic)
	res, err := req.Bytes()
	if err != nil {
		err = fmt.Errorf("NSQ：lookup request error %q", err)
		return
	}

	lu := nsqLookup{}
	err = json.Unmarshal(res, &lu)
	if err != nil {
		err = fmt.Errorf("NSQ: lookup response error %q", err)
		return
	}

	if lu.Status_code == 200 {
		n := len(lu.Data.Producers)

		if n > 0 {
			t := time.Now().Unix()
			i := t % int64(n)

			for index, v := range lu.Data.Producers {
				if int64(index) == i {
					addr = fmt.Sprintf("%s:%v", v.Broadcast_address, v.Tcp_port)
					return
				}
			}
		}

		err = fmt.Errorf("NSQ：%q nsqd service not found", p.topic)
	} else {
		err = fmt.Errorf("NSQ：lookup error %q", lu.Status_txt)
	}

	return
}

type mqNsqConsumer struct {
	isInit   bool
	isHandle bool
	topic    string
	channel  string
	consumer *nsq.Consumer
}

func (c *mqNsqConsumer) Init(topic, channel string) (err error) {
	if len(topic) == 0 {
		err = errors.New(`NSQ: Consumer Init error "topic cannot be empty"`)
		return
	}

	if len(channel) == 0 {
		err = errors.New(`NSQ: Consumer Init error "channel cannot be empty"`)
		return
	}

	nconf := nsq.NewConfig()
	nc, err := nsq.NewConsumer(topic, channel, nconf)
	if err != nil {
		err = fmt.Errorf("NSQ: Consumer Init error %q", err)
		return
	}

	nc.SetLogger(log.Logger, nsq.LogLevelWarning)
	c.topic = topic
	c.channel = channel
	c.consumer = nc
	c.isInit = true
	return
}

func (c *mqNsqConsumer) Handler(chi ConsumerHandlerInterface) (err error) {
	if !c.isInit {
		err = errors.New(`NSQ: Consumer Handler error "must first call the Init func"`)
		return
	}

	if c.isHandle {
		err = errors.New(`NSQ: Consumer Handler error "Handler is running, cannot call it again"`)
		return
	}

	c.consumer.AddHandler(nsq.HandlerFunc(func(m *nsq.Message) (err error) {
		e := errors.NewRecover()
		err = e
		defer mqs.RecoverConsumerPanic(e)

		// handle the message
		err = chi.HandlerMessage(m.Body)
		if err != nil {
			log.Logger.Error("消息处理状态：", err)
		}

		return
	}))

	lookupAddr := fmt.Sprintf("%s:%v", config.Configer.ApiConf.MqServer, config.Configer.ApiConf.MqPort)
	err = c.consumer.ConnectToNSQLookupd(lookupAddr)
	if err != nil {
		err = fmt.Errorf("NSQ: Consumer Handler error %q", err)
	}

	c.isHandle = true
	return
}

func (c *mqNsqConsumer) Stop() (err error) {
	if !c.isInit {
		err = errors.New(`NSQ：Consumer Stop error "must first call the Init func"`)
		return
	}

	c.consumer.Stop()

	// 阻塞等待消息消费完成
	<-c.consumer.StopChan

	return
}

type nsqLookup struct {
	Status_code int
	Status_txt  string
	Data        struct {
		Producers []nsqProducer
	}
}

type nsqProducer struct {
	Broadcast_address string
	Tcp_port          int
	Http_port         int
}
